package org.hscoder.java.opentsdb;

import org.apache.http.nio.reactor.IOReactorException;
import org.hscoder.springboot.simplebuild.util.JsonUtil;
import org.opentsdb.client.OpenTSDBClient;
import org.opentsdb.client.OpenTSDBClientFactory;
import org.opentsdb.client.OpenTSDBConfig;
import org.opentsdb.client.bean.request.*;
import org.opentsdb.client.bean.response.DetailResult;
import org.opentsdb.client.bean.response.LastPointQueryResult;
import org.opentsdb.client.http.callback.BatchPutHttpResponseCallback;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutionException;

public class TsdbTest {

    public static final Logger logger = LoggerFactory.getLogger(TsdbTest.class);

    private static OpenTSDBClient buildClient(String host, int port) throws IOReactorException {
        OpenTSDBConfig config = OpenTSDBConfig.address(host, port)
                // http连接池大小，默认100
                .httpConnectionPool(100)
                // http请求超时时间，默认100s
                .httpConnectTimeout(100)
                // 异步写入数据时，每次http提交的数据条数，默认50
                .batchPutSize(50)
                // 异步写入数据中，内部有一个队列，默认队列大小20000
                .batchPutBufferSize(20000)
                // 异步写入等待时间，如果距离上一次请求超多300ms，且有数据，则直接提交
                .batchPutTimeLimit(300)
                // 当确认这个client只用于查询时设置，可不创建内部队列从而提高效率
                //.readonly()
                // 每批数据提交完成后回调
                .batchPutCallBack(new BatchPutHttpResponseCallback.BatchPutCallBack() {
                    @Override
                    public void response(List<Point> points, DetailResult result) {
                        // 在请求完成并且response code成功时回调
                    }

                    @Override
                    public void responseError(List<Point> points, DetailResult result) {
                        // 在response code失败时回调
                    }

                    @Override
                    public void failed(List<Point> points, Exception e) {
                        // 在发生错误是回调
                    }
                }).config();
        OpenTSDBClient client = OpenTSDBClientFactory.connect(config);
        return client;
    }

    public static void main(String[] args) {
        String host = "http://47.106.80.151";
        int port = 4242;

        String metric = "sys.cpu.nice";

        OpenTSDBClient client = null;
        try {
            logger.info("connect to {}: {}", host, port);
            client = buildClient(host, port);

            long timestamp = System.currentTimeMillis();

            //写入数据
            for (int i = 0; i < 10; i++) {
                long ts = timestamp + (i * 1000);
                double value = Math.random() * 10000;
                Point point = Point.metric(metric)
                        .tag("host", "web01")
                        .value(ts, value)
                        .build();

                logger.info("add point {}: {}", ts, value);
                client.put(point);
            }

            //由于是异步写入，等待2秒钟
            Thread.sleep(2000);

            // 查询数据
            LastPointQuery query = LastPointQuery.sub(LastPointSubQuery.metric(metric).tag("host", "web01").build())
                    .backScan(24)  //过去的24个小时内
                    .build();
            List<LastPointQueryResult> queryResults = client.queryLast(query);
            logger.info("read points results: \n{}", JsonUtil.toPrettyJson(queryResults));

            //删除数据
            Query deleteQuery = Query.begin("1y-ago")
                    .sub(SubQuery.metric(metric)
                            .aggregator(SubQuery.Aggregator.NONE)
                            .build())
                    .build();
            client.delete(deleteQuery);
            logger.info("delete all data.");

        } catch (IOException | ExecutionException | InterruptedException | IllegalAccessException e) {
            e.printStackTrace();
        } finally {
            if (client != null) {
                try {
                    // 优雅关闭连接，会等待所有异步操作完成
                    client.gracefulClose();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
